package al.xc.task;

import al.xc.common.task.TaskThread;
import al.xc.common.util.GsonUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

public class DelayService extends TaskThread {

    public interface IDataCache {
        /**
         * 获取所有数据
         * @param key
         * @return
         */
        Map<String, String> hgetAll(String key);

        /**
         * 设置数据
         * @param key
         * @param field
         * @param vl
         * @return
         */
        Long hset(String key, String field, String vl);

        /**
         * 删除数据
         * @param key
         * @param field
         * @return
         */
        Long hdel(String key, String... field);
    }

    private AtomicInteger start = new AtomicInteger(0);

    private IDataCache cache = null;

    private Integer taskIndex = 0;

    private Map<Integer, DelayTask> taskMap = new HashMap<>();

    private Function function;

    private DelayQueue<Delayed> delayQueue = new DelayQueue<>();

    /**
     * 增加延迟任务
     * @param runnable
     * @param delayMill
     * @return
     */
    public Integer addQueue(Runnable runnable, long delayMill) {
        DelayTask task = newTask(runnable, delayMill);
        synchronized (delayQueue) {
            delayQueue.put(task);
        }
        return task.getIndex();
    }

    /**
     * 增加延迟事件
     * @param key
     * @param event
     * @param data
     * @param delay
     * @return
     */
    public DelayEvent addQueue(String key, String event, String data, long delay) {
        DelayEvent eventInfo = new DelayEvent(key, event, data, delay);
        synchronized (delayQueue) {
            delayQueue.put(eventInfo);
            /**
             * 保存到redis
             */
            cache.hset(eventInfo.getKey(), GsonUtils.toJson(eventInfo), "1");
        }
        return eventInfo;
    }


    /**
     * 移除延迟任务
     * @param taskIndex
     */
    public void remove(Integer taskIndex) {
        DelayTask task;
        synchronized (taskMap) {
            task = taskMap.remove(taskIndex);
        }
        if (null != task){
            delayQueue.remove(task);
        }
    }

    /**
     * 移除延迟事件
     * @param delayed
     */
    public void remove(DelayEvent delayed) {
        delayQueue.remove(delayed);
    }

    protected DelayTask newTask(Runnable runnable, long delayMill) {
        synchronized (taskMap) {
            while (taskMap.containsKey(++taskIndex)) {
            }
            DelayTask task = new DelayTask(delayMill, taskIndex) {
                @Override
                public void run() {
                    runnable.run();
                }
            };
            taskMap.put(taskIndex, task);
            return task;
        }
    }

    /**
     * 初始化
     * @param count
     * @param function
     * @param cache
     * @param redisKey
     */
    public void initialize(int count, Function function, IDataCache cache, String... redisKey) {
        if (!start.compareAndSet(0, 1)) {
            return;
        }
        this.cache = cache;
        this.function = function;
        if (null != this.cache && redisKey.length > 0) {
            for (String single : redisKey) {
                if (null == single) {
                    continue;
                }
                Map<String, String> result = this.cache.hgetAll(single);
                for(String eventInfo : result.keySet()) {
                    DelayEvent event = GsonUtils.fromJson(eventInfo, DelayEvent.class);
                    delayQueue.put(event);
                }
            }
        }

        initialize(count);
        addTask(this::workEntry);
    }

    public void unInitialize() {
        if (!start.compareAndSet(1, 2)) {
            return;
        }
        unInitialize();
        start.compareAndSet(2, 0);
    }

    public void workEntry() {
        while(1 == start.get()) {
            try {
                Delayed delayed = delayQueue.poll(1000, TimeUnit.MILLISECONDS);
                if (delayed instanceof DelayEvent) {
                    DelayEvent event = (DelayEvent)delayed;
                    this.cache.hdel(event.getKey(), GsonUtils.toJson(event));
                    function.apply(delayed);
                } else if (delayed instanceof DelayTask) {
                    DelayTask task = (DelayTask)delayed;
                    synchronized (taskMap) {
                        taskMap.remove(task.getIndex());
                    }
                    task.run();
                }
            } catch (Exception ex) {}
        }
    }
}
